package cn.tianyigps.obd.command;

import cn.tianyigps.obd.beans.Command;
import cn.tianyigps.obd.cache.ChannelCache;
import cn.tianyigps.obd.template.KafkaTemplate;
import cn.tianyigps.obd.utils.MessageCoding;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.util.Arrays;
import java.util.Optional;

/**
 * Created with IDEA
 *
 * @description: 指令监听
 * @author: cui.guibin
 * @create: 2019/4/10 15:40
 **/

public class CommandConsumer  implements Runnable{

    private static ChannelCache channelCache = ChannelCache.getInstance();
    private static Logger log = LogManager.getLogger(CommandConsumer.class);

    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = KafkaTemplate.consumer();
        consumer.subscribe(Arrays.asList("obd_command"));
        ObjectMapper objectMapper = new ObjectMapper();
        while (true){
            ConsumerRecords<String, String> consumerRecords = consumer.poll(1);
            if (null != consumerRecords && consumerRecords.count() > 0){
                for (ConsumerRecord consumerRecord:consumerRecords) {
                    String value = (String) consumerRecord.value();
                    Command command = null;
                    try {
                        command = objectMapper.readValue(value, Command.class);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    sendCmd(command);
                }
            }
        }
    }

    private static void sendCmd(Command command) {
        if (!Optional.ofNullable(command).isPresent())
            return ;
        ChannelHandlerContext ctx = (ChannelHandlerContext) channelCache.get(command.getEquipmentId());
        if (!Optional.ofNullable(ctx).isPresent())
            return ;
        byte[] bytes = MessageCoding.hexStringToBytes(command.getCmd());
        ctx.writeAndFlush(Unpooled.copiedBuffer(bytes));
        log.info("设备： " + command.getEquipmentId() + "  下发指令为：" + command.getCmd());
    }

    @Override
    public void run() {

    }
}
